package com.shujia.flink.state

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo2ValueState {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val kvDS: DataStream[(String, Long)] = linesDS.flatMap(_.split(",")).map((_, 1L))

    kvDS
      .keyBy(_._1)
      .process(new KeyedProcessFunction[String, (String, Long), (String, Long)] {

        /**
          * 在processElement之前执行，在这个方法中定义状态
          *
          */

        /**
          * ValueState: 单值状态，在flink执行过程中，状态中保存的数据会被checkpoint, 会定时持久化到hdfs中
          *
          * 如果任务失败重启，flink可以读取之前持久化的状态数据，接着计算
          *
          * 由于前面做了keyby 所以每一个key都会有一个状态
          */

        var valueState: ValueState[Long] = _

        override def open(parameters: Configuration): Unit = {
          println("初始化状态")

          //获取flink的环境
          val context: RuntimeContext = getRuntimeContext

          //创建状态的描述对象
          val valueStateDesc = new ValueStateDescriptor[Long]("count", classOf[Long])

          //定义状态
          valueState = context.getState(valueStateDesc)

        }

        /**
          *
          * 每一条数据执行一次
          *
          * @param kv  :一行数据
          * @param ctx : 上下文对象，可以获取flink执行环境信息，比如处理时间，事件时间，水位线
          * @param out ：用于将数据发送到下游
          */
        override def processElement(kv: (String, Long), ctx: KeyedProcessFunction[String, (String, Long), (String, Long)]#Context, out: Collector[(String, Long)]): Unit = {
          println("处理数据")
          //获取状态中保存的数据
          //之前统计的结果
          val count: Long = valueState.value()

          //计算新的结果
          val newCount: Long = count + kv._2

          //更新状态
          valueState.update(newCount)

          //将新的单词的数量发送到下游
          out.collect((kv._1, newCount))


        }
      }).print()


    env.execute()
  }

}
